package Services.RocketMqProducer;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import org.apache.thrift.TException;
import org.apache.log4j.Logger;

public class RocketMqProducerImpl implements RocketMqProducer.Iface{



    private Producer producer1; //普通消息生产者
    private OrderProducer producer2; //顺序消息生产者
    private Message message; // 消息
    private static Logger logger = Logger.getLogger(RocketMqProducerImpl.class);

    public RocketMqProducerImpl(Message message,Producer producer1, OrderProducer producer2) {
        this.message = message;
        this.producer1 = producer1;
        this.producer2 = producer2;
    }

    @Override
    public Result sendMsg(CommonMessage message) throws TException{
        logger.info(message);
        this.message.setTopic(message.topic);
        this.message.setTag(message.tags);
        this.message.setBody(message.msg.getBytes());
        if (message.isSetKey()) {
            this.message.setKey(message.key);
        }
        SendResult sendResult = this.producer1.send(this.message);
        // 同步发送消息，只要不抛异常就是成功
        Result result = new Result(0, "ok");
        if (sendResult != null) {
            result.result = sendResult.getMessageId();
            logger.info(sendResult.getMessageId());
        }
        return result;
    }

    @Override
    public Result sendDelayMsg(DelayMessage message) throws TException{
        logger.info(message);
        this.message.setTopic(message.topic);
        this.message.setTag(message.tags);
        this.message.setBody(message.msg.getBytes());
        if (message.isSetSeconds()) {
            this.message.setStartDeliverTime(System.currentTimeMillis() + message.seconds * 1000);
        }
        if (message.isSetKey()) {
            this.message.setKey(message.key);
        }
        SendResult sendResult = this.producer1.send(this.message);
        // 同步发送消息，只要不抛异常就是成功
        Result result = new Result(0, "ok");
        if (sendResult != null) {
            result.result = sendResult.getMessageId();
            logger.info(sendResult.getMessageId());
        }
        return result;
    }

    @Override
    public Result sendOrderMsg(OrderMessage message) throws TException{
        logger.info(message);
        this.message.setTopic(message.topic);
        this.message.setTag(message.tags);
        this.message.setBody(message.msg.getBytes());
        if (message.isSetKey()) {
            this.message.setKey(message.key);
        }
        SendResult sendResult = this.producer2.send(this.message, message.shardingKey);
        // 同步发送消息，只要不抛异常就是成功
        Result result = new Result(0, "ok");
        if (sendResult != null) {
            result.result = sendResult.getMessageId();
            logger.info(sendResult.getMessageId());
        }
        return result;
    }
}
